package org.swift.mashup.engine.concurrency.agenda;

import java.util.ArrayList;
import java.util.Collection;
import java.util.List;
import java.util.concurrent.Callable;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.ScheduledFuture;
import java.util.concurrent.TimeUnit;

import org.apache.commons.lang.StringUtils;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;

@SuppressWarnings("unchecked")
public class TaskManager {

	public static final String POOL_SIZE_PROPERTY = TaskManager.class.getName()
		+ ".thread_pool_size";
	public static final String TIMEOUT_POOL_SIZE_PROPERTY = TaskManager.class.getName()
	+ ".timeout_thread_pool_size";
	
	private static final Log log = LogFactory.getLog(TaskManager.class);
	private static volatile TaskManager _instance = null;
	
	private ExecutorService executor = getExecutorService();
	private ScheduledExecutorService timeoutExecutor = getScheculedExecutorService();
	
	public static TaskManager getInstance(){
		if (_instance == null){
			synchronized (TaskManager.class){
				if (_instance == null)
					_instance = new TaskManager();
			}
		}
		return _instance;
	}
	
	public Collection<Future> executeTasksAsync(Collection<Callable> tasks){
		List<Future> futures = new ArrayList<Future>();
		for (Callable task : tasks){
			futures.add(execute(task));
		}
		return futures;
	}
	
	public <T> Future<T> execute (Callable<T> task){
		return executor.submit(task);
	}
	
	public Future<?> execute (Runnable runnable){
		return executor.submit(runnable);
	}
	
	public <T> Future<T> execute (Runnable runnable, T result){
		return executor.submit(runnable,result);
	}

	public ScheduledFuture<?> scheduleTimeout(ConcurrentAgenda.TimeoutTask timeoutTask, 
			long delay, TimeUnit unit) {
		return this.timeoutExecutor.schedule(timeoutTask, delay, unit);
	}

	private static ExecutorService getExecutorService() {
		ExecutorService service = null;
		final String sizeProp = System.getProperty(POOL_SIZE_PROPERTY);

		if(!StringUtils.isEmpty(sizeProp)) {
			try {
				final int size = Integer.parseInt(sizeProp);
				log.info("Creating thread pool of size " + size);
				service = Executors.newFixedThreadPool(size);
			}
			catch(Exception e) {
				log.error("Couldn't created thread pool with "
						+ "invalid size " + sizeProp, e);
				e.printStackTrace();
			}
		}

		if(service == null) {
			log.info("Creating unbounded thread pool");
			service = Executors.newCachedThreadPool();
		}

		return service;
	}

	private static ScheduledExecutorService getScheculedExecutorService() {
		ScheduledExecutorService service = null;
		int size = 2;
		final String sizeProp = System.getProperty(TIMEOUT_POOL_SIZE_PROPERTY);
		if(!StringUtils.isEmpty(sizeProp)) {
			try {
				size = Integer.parseInt(sizeProp);
			} catch (NumberFormatException e) {
				log.warn("Attempting to use invalid int value for timeout thread pool size: " + 
						sizeProp + " , defaulting to " + size);
			} catch (Exception e) {
				log.error("Error occurred when trying to set timeout thread pool size with value: " + 
						sizeProp + " , defaulting to " + size);
			}
		}

		try {
			log.info("Creating thread pool of size " + size);
			service = Executors.newScheduledThreadPool(size);
		}
		catch(Exception e) {
			log.error("Couldn't created thread pool with "
					+ "invalid size " + sizeProp, e);
			e.printStackTrace();
		}

		return service;
	}
}
